package com.acquisition.task;


import android.serialport.SerialPort;


import com.acquisition.entity.RealTimeData;
import com.acquisition.task.constants.Hj212Constants;
import com.alibaba.fastjson.JSON;

import com.acquisition.task.constants.Constants;
import com.acquisition.task.service.Hj212Service;
import com.acquisition.task.domain.Result;
import com.acquisition.task.service.impl.Hj212ServiceImpl;
import com.acquisition.task.util.ByteUtils;

import com.acquisition.entity.RealTimeData;
import com.acquisition.entity.SerialData;
import com.acquisition.service.DataCollectionService;
import com.acquisition.service.MinuteDataService;
import com.acquisition.service.RealTimeDataService;
import com.acquisition.service.SerialDataService;

import com.acquisition.service.impl.DataCollectionServiceImpl;
import com.acquisition.service.impl.MinuteDataServiceImpl;
import com.acquisition.service.impl.RealTimeDataServiceImpl;
import com.acquisition.service.impl.SerialDataServiceImpl;


import com.acquisition.param.domain.MonitorFactorParam;
import com.acquisition.param.domain.SerialPortParam;
import com.acquisition.param.domain.SingleSerialPortParam;
import com.acquisition.param.service.ParamService;
import com.acquisition.param.service.impl.ParamServiceImpl;



//import com.dcs.cn.common.constants.Constants;
//import com.dcs.cn.common.domain.Result;
//import com.dcs.cn.common.service.Hj212Service;
//import com.dcs.cn.common.service.impl.Hj212ServiceImpl;
//import com.dcs.cn.common.util.ByteUtils;

//import com.dcs.cn.data.domain.RealTimeData;
//import com.dcs.cn.data.domain.SerialData;
//import com.dcs.cn.data.service.DataCollectionService;
//import com.dcs.cn.data.service.MinuteDataService;
//import com.dcs.cn.data.service.RealTimeDataService;
//import com.dcs.cn.data.service.SerialDataService;

//import com.dcs.cn.data.service.impl.DataCollectionServiceImpl;
//import com.dcs.cn.data.service.impl.MinuteDataServiceImpl;
//import com.dcs.cn.data.service.impl.RealTimeDataServiceImpl;
//import com.dcs.cn.data.service.impl.SerialDataServiceImpl;

//import com.dcs.cn.param.domain.MonitorFactorParam;
//import com.dcs.cn.param.domain.SerialPortParam;
//import com.dcs.cn.param.domain.SingleSerialPortParam;
//import com.dcs.cn.param.service.ParamService;
//import com.dcs.cn.param.service.impl.ParamServiceImpl;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedInputStream;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 采集任务
 *
 * @author wuqiwei
 * @date 2021/06/15
 */
public class CollectionTask {

    private static final Logger LOG = LoggerFactory.getLogger(CollectionTask.class);

    public static RealTimeData realTimeData = null;

    /**
     * 数据采集任务
     *
     * @param scheduledThreadPoolExecutor 线程池
     */
    public void run(ScheduledThreadPoolExecutor scheduledThreadPoolExecutor) {
        ParamService paramService = (ParamService) new ParamServiceImpl();
        DataCollectionService dataCollectionService = new DataCollectionServiceImpl();

        // 查询串口参数配置 key=SerialPortParam
        String serialPortParamValue = paramService.getValue(Constants.PARAM_KEY_SERIALPORT);
        if (serialPortParamValue == null || "".equals(serialPortParamValue)) {
            LOG.error("串口参数为空，Param表里key：SerialPortParam，对应的value为空");
            return;
        }
        // 查询监测因子参数配置  key= MonitorFactorParam
        String monitorFactorParamValue = paramService.getValue(Constants.PARAM_KEY_MONITOR);
        if (monitorFactorParamValue == null || "".equals(monitorFactorParamValue)) {
            LOG.error("监测因子参数为空，Param表里key：MonitorFactorParam，对应的value为空");
            return;
        }
        SerialPortParam serialPortParam = JSON.parseObject(serialPortParamValue, SerialPortParam.class);
        MonitorFactorParam monitorFactorParam = JSON.parseObject(monitorFactorParamValue, MonitorFactorParam.class);
        if (serialPortParam == null || monitorFactorParam == null || monitorFactorParam.getInterval() == null || monitorFactorParam.getInterval() == 0) {
            LOG.error("SerialPortParam为空或MonitorFactorParam为空或采集间隔为空或0");
            return;
        }
        int i = 0;
        List<List<String>> list = serialHandle(monitorFactorParam);
        for (String portCode : list.get(0)) {
            // hj212-2017版串口采集
            hj2017SerialDataCollection(scheduledThreadPoolExecutor, 1000L * i++, Objects.requireNonNull(serialPortParam.getSerialPortMap().get(portCode)));
        }
        for (String portCode : list.get(1)) {
            // hj212-2005版串口采集
            hj2005SerialDataCollection(scheduledThreadPoolExecutor, 1000L * i++, Objects.requireNonNull(serialPortParam.getSerialPortMap().get(portCode)));
        }
        // 采集实时数据一分钟一条
        collectionRealTimeData(scheduledThreadPoolExecutor, monitorFactorParam);
        // 采集分钟数据十分钟一条
        collectionMinuteData(scheduledThreadPoolExecutor);
        // 采集小时数据一小时一条
        collectionHourData(scheduledThreadPoolExecutor);
        // 采集日数据一天一条
        collectionDayData(scheduledThreadPoolExecutor);
    }

    /**
     * 串口处理，找出接收2017，2005版协议的串口
     *
     * @param monitorFactorParam MonitorFactorParam
     * @return List<List<String>> get(0):2017版协议的串口, get(1):2005版协议的串口, 其中存储的是"监测因子key,协议类型"
     */
    private List<List<String>> serialHandle(MonitorFactorParam monitorFactorParam) {
        List<List<String>> list = new ArrayList<>();
        List<String> portCode2017 = new ArrayList<>();
        List<String> portCode2005 = new ArrayList<>();
        monitorFactorParam.getMonitorFactorMap().forEach((k, v) -> {
            // 把特殊串口过滤出来，该串口含有多条监测因子的数据，单独进行采集
            if (v != null && v.getMonitorFactorKey() != null && !"".equals(v.getMonitorFactorKey()) && v.getPortCode() != null && !"".equals(v.getPortCode())) {
                if (v.getProtocolType() == Constants.PROTOCOL_TYPE_2017) {
                    portCode2017.add(v.getPortCode());
                } else if (v.getProtocolType() == Constants.PROTOCOL_TYPE_2005) {
                    portCode2005.add(v.getPortCode());
                }
            }
        });
        // 去重
        list.add(portCode2017.stream().distinct().collect(Collectors.toList()));
        list.add(portCode2005.stream().distinct().collect(Collectors.toList()));
        return list;
    }

    /**
     * hj212-2017版串口采集
     *
     * @param scheduledThreadPoolExecutor 线程池
     * @param delay 启动线程的延迟
     * @param singleSerialPortParam SingleSerialPortParam
     */
    @SuppressWarnings({"InfiniteLoopStatement", "BusyWait"})
    private void hj2017SerialDataCollection(ScheduledThreadPoolExecutor scheduledThreadPoolExecutor, long delay, SingleSerialPortParam singleSerialPortParam) {
        scheduledThreadPoolExecutor.schedule(() -> {
            Calendar calendars = Calendar.getInstance();
            SerialDataService serialDataService = new SerialDataServiceImpl();
            Hj212Service hj212Service = new Hj212ServiceImpl();
            SerialPort serialPort = SerialPort.newBuilder(singleSerialPortParam.getPortCode(), singleSerialPortParam.getBaudRate())
                    .dataBits(singleSerialPortParam.getDataBits())
                    .parity(singleSerialPortParam.getParity())
                    .stopBits(singleSerialPortParam.getStopBits())
                    .build();
            InputStream inputStream = new BufferedInputStream(serialPort.getInputStream());
            while (true) {
                if (inputStream.available() > 0) {
                    Thread.sleep(500);
                    byte[] buffer = new byte[inputStream.available()];
                    int size = inputStream.read(buffer);
                    String msg = new String(buffer, StandardCharsets.UTF_8);
                    if (msg.length() > 0) {
                        try {
                            String[] temp = msg.split(Hj212Constants.PACKAGE_HEAD);
                            if (temp.length - 2 <= 0) {
                                msg = Hj212Constants.PACKAGE_HEAD + temp[temp.length - 1];
                            } else {
                                msg = Hj212Constants.PACKAGE_HEAD + temp[temp.length - 2];
                            }
                            LOG.info("该串口：{}，接收到的数据：{}", singleSerialPortParam.getPortCode(), msg);
                            Map<String, String> map = hj212Service.hj2017To2017Map(msg);
                            Map<String, Object> params = new HashMap<>(16);
                            params.put("AND_EQ_portCode", singleSerialPortParam.getPortCode());
                            int count;
                            List<SerialData> list = serialDataService.queryByExample(params);
                            if (list == null || list.size() <= 0) {
                                count = serialDataService.save(new SerialData(null, Long.valueOf(Objects.requireNonNull(map.get(Hj212Constants.HJ212_DATATIME))), Constants.PROTOCOL_TYPE_2017, singleSerialPortParam.getPortCode(), map.get("monitorFactorKey"), map.get("data")));
                            } else {
                                SerialData serialData = list.get(0);
                                serialData.setDataTime(Long.valueOf(Objects.requireNonNull(map.get(Hj212Constants.HJ212_DATATIME))));
                                serialData.setType(Constants.PROTOCOL_TYPE_2017);
                                serialData.setMonitorFactorKey(map.get("monitorFactorKey"));
                                serialData.setData(map.get("data"));
                                count = serialDataService.update(serialData);
                            }
                            int second = calendars.get(Calendar.SECOND);
                            second = second < 30 ? 20000 : 10000;
                            if (count > 0) {
                                LOG.info("{}，串口数据，保存成功，线程休眠{}ms", singleSerialPortParam.getPortCode(), second);
                            } else {
                                LOG.error("{}，串口数据，保存失败，接收的报文：{}，线程休眠{}ms", singleSerialPortParam.getPortCode(), msg, second);
                            }
                            Thread.sleep(second);
                        } catch (Exception e) {
                            e.printStackTrace();
                            LOG.error("{}，串口数据，解析异常，接收的报文：{}", singleSerialPortParam.getPortCode(), msg);
                        }
                    }
                }
            }
        }, delay, TimeUnit.MILLISECONDS);
    }

    /**
     * hj212-2005版串口采集
     *
     * @param scheduledThreadPoolExecutor 线程池
     * @param delay 启动线程的延迟
     * @param singleSerialPortParam SingleSerialPortParam
     */
    @SuppressWarnings({"InfiniteLoopStatement", "BusyWait"})
    private void hj2005SerialDataCollection(ScheduledThreadPoolExecutor scheduledThreadPoolExecutor, long delay, SingleSerialPortParam singleSerialPortParam) {
        scheduledThreadPoolExecutor.schedule(() -> {
            Calendar calendars = Calendar.getInstance();
            SerialDataService serialDataService = new SerialDataServiceImpl();
            Hj212Service hj212Service = new Hj212ServiceImpl();
            SerialPort serialPort = SerialPort.newBuilder(singleSerialPortParam.getPortCode(), singleSerialPortParam.getBaudRate())
                    .dataBits(singleSerialPortParam.getDataBits())
                    .parity(singleSerialPortParam.getParity())
                    .stopBits(singleSerialPortParam.getStopBits())
                    .build();
            InputStream inputStream = new BufferedInputStream(serialPort.getInputStream());
            while (true) {
                if (inputStream.available() > 0) {
                    Thread.sleep(500);
                    byte[] buffer = new byte[inputStream.available()];
                    int size = inputStream.read(buffer);
                    String msg = new String(buffer, StandardCharsets.UTF_8);
                    try {
                        String[] temp = msg.split(Hj212Constants.PACKAGE_HEAD);
                        if (temp.length - 2 <= 0) {
                            msg = Hj212Constants.PACKAGE_HEAD + temp[temp.length - 1];
                        } else {
                            msg = Hj212Constants.PACKAGE_HEAD + temp[temp.length - 2];
                        }
                        LOG.info("该串口：{}，接收到的数据：{}", singleSerialPortParam.getPortCode(), msg);
                        Map<String, String> map = hj212Service.hj2005To2017Map(msg);
                        Map<String, Object> params = new HashMap<>(16);
                        params.put("AND_EQ_portCode", singleSerialPortParam.getPortCode());
                        int count;
                        List<SerialData> list = serialDataService.queryByExample(params);
                        if (list == null || list.size() <= 0) {
                            count = serialDataService.save(new SerialData(null, Long.valueOf(Objects.requireNonNull(map.get(Hj212Constants.HJ212_DATATIME))), Constants.PROTOCOL_TYPE_2005, singleSerialPortParam.getPortCode(), map.get("monitorFactorKey"), map.get("data")));
                        } else {
                            SerialData serialData = list.get(0);
                            serialData.setDataTime(Long.valueOf(Objects.requireNonNull(map.get(Hj212Constants.HJ212_DATATIME))));
                            serialData.setType(Constants.PROTOCOL_TYPE_2005);
                            serialData.setMonitorFactorKey(map.get("monitorFactorKey"));
                            serialData.setData(map.get("data"));
                            count = serialDataService.update(serialData);
                        }
                        int second = calendars.get(Calendar.SECOND);
                        second = second < 30 ? 20000 : 10000;
                        if (count > 0) {
                            LOG.info("{}，串口数据，保存成功，线程休眠{}ms", singleSerialPortParam.getPortCode(), second);
                        } else {
                            LOG.error("{}，串口数据，保存失败，接收的报文：{}，线程休眠{}ms", singleSerialPortParam.getPortCode(), msg, second);
                        }
                        Thread.sleep(second);
                    } catch (Exception e) {
                        e.printStackTrace();
                        LOG.error("{}，串口数据，解析异常，接收的报文：{}", singleSerialPortParam.getPortCode(), msg);
                    }
                }
            }
        }, delay, TimeUnit.MILLISECONDS);
    }

    /**
     * 河北省协议版串口采集
     *
     * @param scheduledThreadPoolExecutor 线程池
     * @param delay 启动线程的延迟
     * @param singleSerialPortParam SingleSerialPortParam
     */
    @SuppressWarnings("BusyWait")
    private void hebeiSerialDataCollection(ScheduledThreadPoolExecutor scheduledThreadPoolExecutor, long delay, SingleSerialPortParam singleSerialPortParam) {
        SerialDataService serialDataService = new SerialDataServiceImpl();
        Hj212Service hj212Service = new Hj212ServiceImpl();
        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
            try {
                SerialPort serialPort = SerialPort.newBuilder(singleSerialPortParam.getPortCode(), singleSerialPortParam.getBaudRate())
                        .dataBits(singleSerialPortParam.getDataBits())
                        .parity(singleSerialPortParam.getParity())
                        .stopBits(singleSerialPortParam.getStopBits())
                        .build();
                long start = System.currentTimeMillis();
                outerLoop : for (int i = 0; i < 3; i++) {
                    // 发送数据
                    serialPort.getOutputStream().write(ByteUtils.hexStrToByteArr("0x23,0x23,0x01,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0x11,0x22,0x26,0x26".replaceAll(",", "").replaceAll("0x", "")));
                    Thread.sleep(500);
                    // 读取数据
                    InputStream inputStream = new BufferedInputStream(serialPort.getInputStream());
                    while (true) {
                        if (inputStream.available() > 0) {
                            Thread.sleep(500);
                            byte[] buffer = new byte[inputStream.available()];
                            int size = inputStream.read(buffer);
                            String msg = ByteUtils.byteArrToHexStr(buffer);
                            LOG.info("该串口：{}，接收到河北省协议报文：{}", singleSerialPortParam.getPortCode(), msg);
                            /*
                             * 初步校验：
                             * 1.报文以2323为包头，2626为包尾
                             * 2.校验数据包长度：数据包长度 = (报文总长度 / 2) - 包头位(2) - 包尾位(2) - 数据包长度位(2)
                             */
                            if (!msg.startsWith("2323") || !msg.endsWith("2626")) {
                                LOG.error("{}，河北省协议，串口数据，包头包尾校验失败，再次重新读取，第{}次，接收的报文：{}", singleSerialPortParam.getPortCode(), i, msg);
                                // 校验失败，再重新读取，开始时间重置
                                start = System.currentTimeMillis();
                                break;
                            }
                            if (Integer.parseInt(msg.substring(4, 8), 16) == msg.length() / 2 - 6) {
                                LOG.error("{}，河北省协议，串口数据，数据包长度校验失败，再次重新读取，第{}次，报文中数据包长度：{}，计算出的数据包长度（报文总长度除2再减6）：{}，接收的报文：{}", singleSerialPortParam.getPortCode(), i, Integer.parseInt(msg.substring(4, 8), 16), msg.length() / 2 - 6, msg);
                                // 校验失败，再重新读取，开始时间重置
                                start = System.currentTimeMillis();
                                break;
                            }
                            Map<String, String> map = hj212Service.hebeiTo2017Map(msg);
                            LOG.info("{}，河北省协议，解析完的串口数据：{}", singleSerialPortParam.getPortCode(), JSON.toJSONString(map));
                            Map<String, Object> params = new HashMap<>(16);
                            params.put("AND_EQ_portCode", singleSerialPortParam.getPortCode());
                            int count;
                            List<SerialData> list = serialDataService.queryByExample(params);
                            if (list == null || list.size() <= 0) {
                                count = serialDataService.save(new SerialData(null, Long.valueOf(Objects.requireNonNull(map.get(Hj212Constants.HJ212_DATATIME))), Constants.PROTOCOL_TYPE_HEBEI, singleSerialPortParam.getPortCode(), map.get("monitorFactorKey"), map.get("data")));
                            } else {
                                SerialData serialData = list.get(0);
                                serialData.setDataTime(Long.valueOf(Objects.requireNonNull(map.get(Hj212Constants.HJ212_DATATIME))));
                                serialData.setType(Constants.PROTOCOL_TYPE_HEBEI);
                                serialData.setMonitorFactorKey(map.get("monitorFactorKey"));
                                serialData.setData(map.get("data"));
                                count = serialDataService.update(serialData);
                            }
                            if (count > 0) {
                                LOG.info("{}，河北省协议，串口数据，保存成功", singleSerialPortParam.getPortCode());
                                // 保存成功，串口关闭，直接跳出外循环
                                serialPort.close();
                                break outerLoop;
                            } else {
                                LOG.error("{}，河北省协议，串口数据，保存失败，再次重新读取，第{}次，接收的报文：{}", singleSerialPortParam.getPortCode(), i, msg);
                                // 保存失败，再重新读取，开始时间重置
                                start = System.currentTimeMillis();
                                break;
                            }
                        }
                        if (System.currentTimeMillis() - start > 5000) {
                            LOG.error("该串口：{}，读取河北省协议数据超时，再次重新读取，第{}次", singleSerialPortParam.getPortCode(), i);
                            // 读取超时，再重新读取，开始时间重置
                            start = System.currentTimeMillis();
                            break;
                        }
                    }
                    // 如果重试三次，还是没有读取成功，则置空该串口的历史数据
                    if (i == 2) {
                        // 判断一下，该串口是否采集过数据了，采集过了才置空
                        Map<String, Object> params = new HashMap<>(16);
                        params.put("AND_EQ_portCode", singleSerialPortParam.getPortCode());
                        List<SerialData> list = serialDataService.queryByExample(params);
                        if (list != null && list.size() > 0) {
                            SerialData serialData = list.get(0);
                            serialData.setType(Constants.PROTOCOL_TYPE_HEBEI);
                            serialData.setMonitorFactorKey("");
                            serialData.setData("");
                            LOG.error("{}，河北省协议，多次读取河北省协议数据超时，置空串口数据", singleSerialPortParam.getPortCode());
                            if (serialDataService.update(serialData) < 0) {
                                LOG.error("{}，河北省协议，串口数据，置空失败", singleSerialPortParam.getPortCode());
                            }
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, delay, 60 * 1000, TimeUnit.MILLISECONDS);
    }

    /**
     * 特殊串口，该串口含有多条监测因子的数据，单独进行采集
     *
     * @param scheduledThreadPoolExecutor 线程池
     * @param delay 启动线程的延迟
     * @param singleSerialPortParam SingleSerialPortParam
     * @param specialList 特殊串口，其中存储的是"监测因子key,协议类型"
     */
    @SuppressWarnings({"InfiniteLoopStatement", "BusyWait"})
    private void specialSerialDataCollection(ScheduledThreadPoolExecutor scheduledThreadPoolExecutor, long delay, SingleSerialPortParam singleSerialPortParam, List<String> specialList) {
        scheduledThreadPoolExecutor.schedule(() -> {
            // 存放2017协议的监测因子key值
            List<String> key2017 = new ArrayList<>();
            specialList.forEach(s -> {
                String[] strings = s.split(",");
                if (String.valueOf(Constants.PROTOCOL_TYPE_2017).equals(strings[1])) {
                    key2017.add(strings[0]);
                }
            });
            Calendar calendars = Calendar.getInstance();
            SerialDataService serialDataService = new SerialDataServiceImpl();
            Hj212Service hj212Service = new Hj212ServiceImpl();
            SerialPort serialPort = SerialPort.newBuilder(singleSerialPortParam.getPortCode(), singleSerialPortParam.getBaudRate())
                    .dataBits(singleSerialPortParam.getDataBits())
                    .parity(singleSerialPortParam.getParity())
                    .stopBits(singleSerialPortParam.getStopBits())
                    .build();
            InputStream inputStream = new BufferedInputStream(serialPort.getInputStream());
            while (true) {
                if (inputStream.available() > 0) {
                    Thread.sleep(1000);
                    byte[] buffer = new byte[inputStream.available()];
                    int size = inputStream.read(buffer);
                    String msg = new String(buffer, StandardCharsets.UTF_8);
                    try {
                        int count = 0;
                        String[] temp = msg.split(Hj212Constants.PACKAGE_HEAD);
                        for (String singleMsg : temp) {
                            if (singleMsg.length() > 0) {
                                singleMsg = Hj212Constants.PACKAGE_HEAD + singleMsg;
                                if (singleMsg.contains("ad1=")) {
                                    // ##ad1=3.30,ad2=0.00,ad3=0.00,ad4=0.00,di1=1,di2=1处理
                                    LOG.info("该特殊串口：{}，接收到的特殊信息：{}", singleSerialPortParam.getPortCode(), singleMsg);
                                    count = 1;
                                } else {
                                    // 用于判断singleMsg是2017协议还是2005协议
                                    boolean flag = false;
                                    for (String key : key2017) {
                                        if (singleMsg.contains(key)) {
                                            flag = true;
                                            break;
                                        }
                                    }
                                    Map<String, String> map;
                                    if (flag) {
                                        LOG.info("该特殊串口：{}，2017协议，接收到的数据：{}", singleSerialPortParam.getPortCode(), singleMsg);
                                        map = hj212Service.hj2017To2017Map(singleMsg);
                                    } else {
                                        LOG.info("该特殊串口：{}，2005协议，接收到的数据：{}", singleSerialPortParam.getPortCode(), singleMsg);
                                        map = hj212Service.hj2005To2017Map(singleMsg);
                                    }
                                    Map<String, Object> params = new HashMap<>(16);
                                    params.put("AND_EQ_portCode", singleSerialPortParam.getPortCode());
                                    params.put("AND_EQ_monitorFactorKey", map.get("monitorFactorKey"));
                                    List<SerialData> list = serialDataService.queryByExample(params);
                                    if (list == null || list.size() <= 0) {
                                        count = serialDataService.save(new SerialData(null, Long.valueOf(Objects.requireNonNull(map.get(Hj212Constants.HJ212_DATATIME))), Constants.PROTOCOL_TYPE_2017, singleSerialPortParam.getPortCode(), map.get("monitorFactorKey"), map.get("data")));
                                    } else {
                                        SerialData serialData = list.get(0);
                                        serialData.setDataTime(Long.valueOf(Objects.requireNonNull(map.get(Hj212Constants.HJ212_DATATIME))));
                                        serialData.setMonitorFactorKey(map.get("monitorFactorKey"));
                                        serialData.setData(map.get("data"));
                                        count = serialDataService.update(serialData);
                                    }
                                }
                            }
                        }
                        int second = calendars.get(Calendar.SECOND);
                        second = second < 30 ? 20000 : 10000;
                        if (count > 0) {
                            LOG.info("{}，串口数据，保存成功，线程休眠{}ms", singleSerialPortParam.getPortCode(), second);
                        } else {
                            LOG.error("{}，串口数据，保存失败，接收的报文：{}，线程休眠{}ms", singleSerialPortParam.getPortCode(), msg, second);
                        }
                        Thread.sleep(second);
                    } catch (Exception e) {
                        e.printStackTrace();
                        LOG.error("{}，串口数据，解析异常，接收的报文：{}", singleSerialPortParam.getPortCode(), msg);
                    }
                }
            }
        }, delay, TimeUnit.MILLISECONDS);
    }

    /**
     * 实时数据采集（一分钟一条，整点）
     *
     * @param scheduledThreadPoolExecutor 线程池
     * @param monitorFactorParam MonitorFactorParam
     */
    private void collectionRealTimeData(ScheduledThreadPoolExecutor scheduledThreadPoolExecutor, MonitorFactorParam monitorFactorParam) {
        Calendar calendar = Calendar.getInstance();
        calendar.set(Calendar.MILLISECOND, 0);
        calendar.set(Calendar.SECOND, 0);
        calendar.add(Calendar.MINUTE, 1);
        // 加了3s延迟
        long initialDelay = calendar.getTimeInMillis() - System.currentTimeMillis() + 3000;
        DataCollectionService dataCollectionService = new DataCollectionServiceImpl();
        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
            try {
                Long a = System.currentTimeMillis();
                Long dateTime = Long.parseLong(new SimpleDateFormat("yyyyMMddHHmm", Locale.getDefault()).format(Calendar.getInstance().getTime()) + "00");
                Result result = dataCollectionService.saveRealTimeData(dateTime);
                String realTimeDataString = result.getData();
                realTimeData = JSON.parseObject(realTimeDataString, RealTimeData.class);
                if (result.getCode() == Constants.RESULT_CODE_SUCCESS) {
                    LOG.info("实时数据采集成功");
                } else {
                    LOG.error("实时数据采集失败，原因：{}", result.getMsg());
                }
                Long b = System.currentTimeMillis();
                LOG.info("当前APP版本：{}，实时数据采集任务运行时长：{}ms\r\n", Constants.APP_VERSION, b - a);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, initialDelay, monitorFactorParam.getInterval() * 1000, TimeUnit.MILLISECONDS);
    }

    /**
     * 分钟数据采集（十分钟一条，整点）
     *
     * @param scheduledThreadPoolExecutor 线程池
     */
    private void collectionMinuteData(ScheduledThreadPoolExecutor scheduledThreadPoolExecutor) {
        Calendar calendars = Calendar.getInstance();
        calendars.set(Calendar.MILLISECOND, 0);
        calendars.set(Calendar.SECOND, 0);
        int minute = calendars.get(Calendar.MINUTE);
        calendars.set(Calendar.MINUTE, ((minute / 10) + 1) * 10);
        // 加了4s延迟
        long initialDelay = calendars.getTimeInMillis() - System.currentTimeMillis() + 4000;
        DataCollectionService dataCollectionService = new DataCollectionServiceImpl();
        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
            try {
                Long a = System.currentTimeMillis();
                Calendar calendar = Calendar.getInstance();
                Long endTime = Long.parseLong(new SimpleDateFormat("yyyyMMddHHmm", Locale.getDefault()).format(calendar.getTime()) + "00");
                calendar.add(Calendar.MINUTE, -10);
                Long beginTime = Long.parseLong(new SimpleDateFormat("yyyyMMddHHmm", Locale.getDefault()).format(calendar.getTime()) + "00");
                Result result = dataCollectionService.saveMinuteData(beginTime, endTime);
                if (result.getCode() == Constants.RESULT_CODE_SUCCESS) {
                    LOG.info("分钟数据采集成功");
                } else {
                    LOG.error("分钟数据采集失败，原因：{}", result.getMsg());
                }
                Long b = System.currentTimeMillis();
                LOG.info("当前APP版本：{}，分钟数据采集任务运行时长：{}ms\r\n", Constants.APP_VERSION, b - a);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, initialDelay, 10 * 60 * 1000, TimeUnit.MILLISECONDS);
    }

    /**
     * 小时数据采集（一小时一条，整点）
     *
     * @param scheduledThreadPoolExecutor 线程池
     */
    private void collectionHourData(ScheduledThreadPoolExecutor scheduledThreadPoolExecutor) {
        Calendar calendars = Calendar.getInstance();
        calendars.set(Calendar.MILLISECOND, 0);
        calendars.set(Calendar.SECOND, 0);
        calendars.set(Calendar.MINUTE, 0);
        calendars.add(Calendar.HOUR, 1);
        // 加了5s延迟
        long initialDelay = calendars.getTimeInMillis() - System.currentTimeMillis() + 5000;
        DataCollectionService dataCollectionService = new DataCollectionServiceImpl();
        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
            try {
                Long a = System.currentTimeMillis();
                Calendar calendar = Calendar.getInstance();
                Long endTime = Long.parseLong(new SimpleDateFormat("yyyyMMddHH", Locale.getDefault()).format(calendar.getTime()) + "0000");
                calendar.add(Calendar.HOUR, -1);
                Long beginTime = Long.parseLong(new SimpleDateFormat("yyyyMMddHH", Locale.getDefault()).format(calendar.getTime()) + "0000");
                Result result = dataCollectionService.saveHourData(beginTime, endTime);
                if (result.getCode() == Constants.RESULT_CODE_SUCCESS) {
                    LOG.info("小时数据采集成功");
                } else {
                    LOG.error("小时数据采集失败，原因：{}", result.getMsg());
                }
                Long b = System.currentTimeMillis();
                LOG.info("当前APP版本：{}，小时数据采集任务运行时长：{}ms\r\n", Constants.APP_VERSION, b - a);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, initialDelay, 60 * 60 * 1000, TimeUnit.MILLISECONDS);
    }

    /**
     * 日数据采集（一天一条，整点）
     *
     * @param scheduledThreadPoolExecutor 线程池
     */
    private void collectionDayData(ScheduledThreadPoolExecutor scheduledThreadPoolExecutor) {
        Calendar calendars = Calendar.getInstance();
        calendars.set(Calendar.MILLISECOND, 0);
        calendars.set(Calendar.SECOND, 0);
        calendars.set(Calendar.MINUTE, 0);
        calendars.set(Calendar.HOUR_OF_DAY, 0);
        calendars.add(Calendar.DAY_OF_MONTH, 1);
        // 加了6s延迟
        long initialDelay = calendars.getTimeInMillis() - System.currentTimeMillis() + 6000;
        DataCollectionService dataCollectionService = new DataCollectionServiceImpl();
        RealTimeDataService realTimeDataService = new RealTimeDataServiceImpl();
        MinuteDataService minuteDataService = new MinuteDataServiceImpl();
        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
            try {
                Long a = System.currentTimeMillis();
                Calendar calendar = Calendar.getInstance();
                Long endTime = Long.parseLong(new SimpleDateFormat("yyyyMMdd", Locale.getDefault()).format(calendar.getTime()) + "000000");
                calendar.add(Calendar.DAY_OF_MONTH, -1);
                Long beginTime = Long.parseLong(new SimpleDateFormat("yyyyMMdd", Locale.getDefault()).format(calendar.getTime()) + "000000");
                Result result = dataCollectionService.saveDayData(beginTime, endTime);
                if (result.getCode() == Constants.RESULT_CODE_SUCCESS) {
                    LOG.info("日数据采集成功");
                } else {
                    LOG.error("日数据采集失败，原因：{}", result.getMsg());
                }
                /*
                 * 删除规则：
                 * 实时数据只保留三天(没上传的也删除)。
                 * 分钟数据只保留三天(没上传的不删除)。
                 * 小时数据永久保存。
                 * 日数据永久保存。
                 */
                Calendar deleteTime = Calendar.getInstance();
                deleteTime.add(Calendar.DAY_OF_MONTH, -3);
                Long deleteTimes = Long.parseLong(new SimpleDateFormat("yyyyMMdd", Locale.getDefault()).format(deleteTime.getTime()) + "000000");
                if (realTimeDataService.remove(deleteTimes) > 0) {
                    LOG.info("实时数据删除成功");
                } else {
                    LOG.error("实时数据删除失败");
                }
                if (minuteDataService.remove(deleteTimes) > 0) {
                    LOG.info("分钟数据删除成功");
                } else {
                    LOG.error("分钟数据删除失败");
                }
                Long b = System.currentTimeMillis();
                LOG.info("当前APP版本：{}，日数据采集任务运行时长：{}ms\r\n", Constants.APP_VERSION, b - a);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, initialDelay, 24 * 60 * 60 * 1000, TimeUnit.MILLISECONDS);
    }

}
